use crate::cmp::{Cmp, InternalKeyCmp};
use crate::env::Env;
use crate::error::{err, Result, StatusCode};
use crate::key_types::{parse_internal_key, InternalKey, UserKey};
use crate::log::{LogReader, LogWriter};
use crate::merging_iter::MergingIter;
use crate::options::Options;
use crate::table_cache::TableCache;
use crate::types::{
    parse_file_name, share, FileMetaData, FileNum, FileType, LdbIterator, Shared, NUM_LEVELS,
};
use crate::version::{new_version_iter, total_size, FileMetaHandle, Version};
use crate::version_edit::VersionEdit;

use bytes::Bytes;
use std::cmp::Ordering;
use std::collections::HashSet;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::rc::Rc;

/// 这段代码定义了 Compaction 结构体，用于描述 LSM 树存储引擎中压缩操作（compaction）的核心参数和状态，是控制压缩过程的关键数据结构。
/// 压缩操作是 LSM 树的核心机制，用于合并多个 SSTable、去除过期数据并优化查询性能，而 Compaction 结构体则封装了该操作的所有必要信息。
pub struct Compaction {
    /// 当前压缩操作的源层级（例如，从 L1 层合并到 L2 层时，level 为 1）。
    level: usize,
    /// 输出 SSTable 的最大文件大小限制，用于控制新生成文件的尺寸。
    max_file_size: usize,
    /// 指向参与压缩的版本（Version）的共享引用，版本记录了当前各层级 SSTable 的状态。
    input_version: Option<Shared<Version>>,
    /// 记录各层级中参与压缩的 SSTable 索引，用于快速定位文件。
    level_ixs: [usize; NUM_LEVELS],
    /// InternalKeyCmp：键比较器，分别用于用户键和内部键（包含序列编号）的排序，确保压缩过程中键的有序性。
    cmp: Rc<Box<dyn Cmp>>,
    /// InternalKeyCmp：键比较器，分别用于用户键和内部键（包含序列编号）的排序，确保压缩过程中键的有序性。
    icmp: InternalKeyCmp,
    /// 标记该压缩操作是手动触发还是自动触发（由存储引擎根据策略自动决定）。
    manual: bool,

    // "parent" inputs from level and level+1.
    /// 存储参与压缩的 SSTable 列表，通常包含两个元素
    /// inputs[0]：源层级（level）中需要压缩的 SSTable。
    /// inputs[1]：下一层级（level+1）中与源层级 SSTable 键范围重叠的 SSTable（需一起合并）。
    inputs: [Vec<FileMetaHandle>; 2],
    /// 记录更高层级（level+2 及以上）中与当前压缩范围重叠的 SSTable，用于判断压缩后的数据是否需要进一步拆分，避免后续压缩成本过高。
    grandparent_ix: usize,
    // remaining inputs from level+2..NUM_LEVELS
    /// 记录更高层级（level+2 及以上）中与当前压缩范围重叠的 SSTable，用于判断压缩后的数据是否需要进一步拆分，避免后续压缩成本过高。
    grandparents: Option<Vec<FileMetaHandle>>,
    /// 记录与更高层级 SSTable 重叠的数据量，用于优化压缩策略（如决定是否拆分输出文件）。
    overlapped_bytes: usize,
    /// 标记是否已处理过键，用于跟踪压缩进度。
    seen_key: bool,
    /// 用于记录压缩操作对版本的修改（如删除旧 SSTable、添加新 SSTable），最终应用于版本管理系统。
    edit: VersionEdit,
}

impl Compaction {
    // Note: opt.cmp should be the user-supplied or default comparator (not an InternalKeyCmp).
    pub fn new(opt: &Options, level: usize, input: Option<Shared<Version>>) -> Compaction {
        Compaction {
            level,
            max_file_size: opt.max_file_size,
            input_version: input,
            level_ixs: Default::default(),
            cmp: opt.cmp.clone(),
            icmp: InternalKeyCmp(opt.cmp.clone()),
            manual: false,

            inputs: Default::default(),
            grandparent_ix: 0,
            grandparents: Default::default(),
            overlapped_bytes: 0,
            seen_key: false,
            edit: VersionEdit::new(),
        }
    }

    fn add_input(&mut self, parent: usize, f: FileMetaHandle) {
        assert!(parent <= 1);
        self.inputs[parent].push(f)
    }

    /// 这段代码定义了 level 方法，它是 Compaction 结构体的一个简单访问器（getter）方法，用于返回当前压缩操作（compaction）的源层级编号。
    ///
    /// 具体来说：
    ///
    /// 方法返回 self.level，即 Compaction 结构体中存储的 level 字段值。
    ///
    /// 这个返回值表示当前压缩操作的源层级（例如，返回 1 表示该压缩操作是从 L1 层发起的，通常会将 L1 层的部分 SSTable 与 L2 层中重叠的 SSTable 进行合并）。
    ///
    /// 在压缩过程中，需要明确知道源层级信息来决定合并策略、目标层级以及文件组织方式，这个方法提供了一种简洁的方式来获取该核心信息。
    ///
    /// 简单来说，它是一个 “层级获取器”，用于快速获取当前压缩操作所属的源层级编号。
    pub fn level(&self) -> usize {
        self.level
    }

    /// 这段代码定义了 input 方法，用于获取压缩操作（Compaction）中指定的输入 SSTable 文件的元数据（FileMetaData），是访问压缩输入文件信息的便捷接口。
    ///
    /// 这个方法的作用是安全地获取压缩操作中特定输入文件的元数据，方便在压缩过程中访问输入 SSTable 的属性（如判断键范围重叠、计算文件大小等）。
    ///
    /// 在压缩逻辑中，需要频繁访问输入文件的元数据来决定合并策略（如如何拆分输出文件、判断是否与更高层级文件重叠等），input 方法提供了统一且安全的访问方式。
    ///
    /// 简单来说，它是压缩操作输入文件的 “元数据访问器”，用于获取指定输入 SSTable 的详细信息。
    pub fn input(&self, parent: usize, ix: usize) -> FileMetaData {
        /// 确保 parent 参数只能是 0 或 1（因为压缩操作的输入通常来自两个层级：源层级 level 和下一层级 level+1，对应 inputs[0] 和 inputs[1]）。
        assert!(parent < 2);
        /// 确保索引 ix 在有效范围内，避免访问超出输入文件列表长度的位置。
        assert!(ix < self.inputs[parent].len());
        /// 通过 self.inputs[parent][ix] 访问指定的输入文件句柄（FileMetaHandle）。
        /// 调用 borrow().clone() 获取该文件的元数据副本（FileMetaData），其中包含文件名、键范围、大小等描述 SSTable 的关键信息。
        self.inputs[parent][ix].borrow().clone()
    }

    /// 这段代码定义了 num_inputs 方法，用于获取压缩操作（Compaction）中指定输入源的 SSTable 文件数量，是访问压缩输入规模的便捷接口。
    ///
    /// 具体功能分析：
    ///
    /// 参数校验：assert!(parent < 2) 确保 parent 参数只能是 0 或 1（因为压缩操作的输入通常来自两个来源：源层级 level 和下一层级 level+1）。
    ///
    /// 返回数量：返回 self.inputs[parent].len()，即指定输入源（parent=0 对应源层级，parent=1 对应下一层级）中参与压缩的 SSTable 文件总数。
    ///
    /// 在压缩过程中，需要知道各输入源的文件数量来判断压缩规模、分配资源或制定合并策略（例如，L0 层文件数量过多时可能触发压缩）。这个方法提供了一种安全的方式来获取这些基础统计信息。
    ///
    /// 简单来说，它是压缩输入的 “文件计数器”，用于快速获取指定来源的参与压缩的 SSTable 数量。
    pub fn num_inputs(&self, parent: usize) -> usize {
        // assert!(parent < 2) 确保 parent 参数只能是 0 或 1（因为压缩操作的输入通常来自两个来源：源层级 level 和下一层级 level+1）。
        assert!(parent < 2);
        // 返回 self.inputs[parent].len()，即指定输入源（parent=0 对应源层级，parent=1 对应下一层级）中参与压缩的 SSTable 文件总数。
        self.inputs[parent].len()
    }

    pub fn edit(&mut self) -> &mut VersionEdit {
        &mut self.edit
    }

    pub fn into_edit(self) -> VersionEdit {
        self.edit
    }

    /// add_input_deletions marks the current input files as deleted in the inner VersionEdit.
    ///
    /// 这段段代码定义了 add_input_deletions 方法，用于在压缩操作（Compaction）中将参与压缩的所有输入 SSTable 文件标记为待删除，并记录到版本编辑（VersionEdit）中，是压缩过程中清理旧文件的关键步骤。
    ///
    /// 在 LSM 树中，压缩操作的目的是合并旧的 SSTable 并生成新的 SSTable。当压缩完成后，参与压缩的旧文件已不再需要（其数据已合并到新文件中），因此需要通过该方法标记为删除。
    /// 这些标记会在版本更新时生效，确保旧文件从存储引擎的活跃版本中移除，最终被物理删除以释放磁盘空间。
    ///
    /// 简单来说，它是压缩过程中的 “旧文件清理标记器”，负责将所有参与压缩的旧 SSTable 标记为待删除，为后续的版本更新和空间回收做准备。
    pub fn add_input_deletions(&mut self) {
        // 遍历输入源：通过双层循环遍历压缩操作的两个输入源（parent in 0..2），即源层级（level）和下一层级（level+1）中参与压缩的所有 SSTable 文件（f in &self.inputs[parent]）。
        for parent in 0..2 {
            for f in &self.inputs[parent] {
                // 标记删除：对每个输入文件，调用 self.edit.delete_file 方法，将其所在层级（self.level + parent）和文件编号（f.borrow().num）记录到版本编辑（self.edit）的待删除集合中。
                self.edit.delete_file(self.level + parent, f.borrow().num);
            }
        }
    }

    /// is_base_level_for checks whether the given key may exist in levels higher than this
    /// compaction's level plus 2. I.e., whether the levels for this compaction are the last ones
    /// to contain the key.
    ///
    /// 这段代码定义了 is_base_level_for 方法，用于判断在 LSM 树压缩（compaction）过程中，当前压缩的层级是否是包含指定用户键（k）的最高层级（即 “基础层级”）。
    /// 这一判断对压缩过程中的数据清理策略至关重要，尤其是决定是否可以安全删除旧版本的键。
    ///
    /// 这个方法的核心作用是支持 LSM 树的 “版本修剪” 策略：在压缩时，对于被标记为基础层级的键，可以安全删除其所有旧版本（因为更高层级没有副本），从而减少存储空间占用。这是优化 LSM 树读写性能的重要机制。
    ///
    /// 简单来说，它是压缩过程中的 “键版本检查器”，用于判断当前压缩层级是否是某个键的最高存储层级，为数据清理提供依据。
    pub fn is_base_level_for(&mut self, k: UserKey<'_>) -> bool {
        assert!(self.input_version.is_some());
        // inp_version 是当前版本的 SSTable 元数据，通过它可以访问各层级的文件列表（files）。
        let inp_version = self.input_version.as_ref().unwrap();
        // 检查指定用户键 k 是否可能存在于当前压缩层级（self.level）+2 及更高的层级中。如果不存在，则当前压缩涉及的层级（self.level 和 self.level+1）就是该键的 “基础层级”，压缩后可以安全清理旧版本；
        // 如果存在，则说明更高层级还有该键的副本，需要保留必要的版本。
        // 从 self.level + 2 开始遍历到最高层级（NUM_LEVELS），检查这些层级中是否有包含键 k 的 SSTable。
        for level in self.level + 2..NUM_LEVELS {
            let files = &inp_version.borrow().files[level];
            // 对每个高层级（level），使用 level_ixs[level] 记录上次检查的文件索引（跨调用的状态缓存，加速后续查找）。
            while self.level_ixs[level] < files.len() {
                let f = files[self.level_ixs[level]].borrow();
                // 对每个文件，通过比较键 k 与文件的最小键（f.smallest）和最大键（f.largest）的用户键部分（通过 parse_internal_key 提取），判断 k 是否在该文件的键范围内：
                // 若 k 小于等于文件的最大用户键且大于等于最小用户键，则说明该文件可能包含 k，返回 false（当前层级不是基础层级）。
                // 若 k 大于文件的最大用户键，则继续检查下一个文件（递增 level_ixs[level]）。
                // 若 k 小于文件的最小用户键，则无需检查该层级的后续文件（因 SSTable 按键排序），跳出循环。
                if self.cmp.cmp(k, parse_internal_key(&f.largest).2) <= Ordering::Equal {
                    if self.cmp.cmp(k, parse_internal_key(&f.smallest).2) >= Ordering::Equal {
                        // key is in this file's range, so this is not the base level.
                        return false;
                    }
                    break;
                }
                // level_ixs contains cross-call state to speed up following lookups.
                self.level_ixs[level] += 1;
            }
        }
        // 若所有更高层级都不包含 k，返回 true（当前层级是基础层级）。
        // 否则返回 false。
        true
    }

    /// 这个 is_trivial_move 方法用于判断当前压缩任务（Compaction）是否可以执行 “简单移动”（trivial move）优化。
    /// 在 LSM 树中，“简单移动” 是一种轻量级操作，当满足特定条件时，可直接将源层级的 SSTable 文件移动到下一层级，无需执行复杂的键值对合并，从而节省大量计算和 I/O 资源。
    ///
    /// is_trivial_move 的核心价值是识别可优化的压缩场景，降低系统开销，主要作用包括：
    ///
    /// 1.减少不必要的计算
    ///
    /// 当满足条件时，直接移动文件而非合并键值对，避免解析、排序、重写键值对的开销（这些操作在大规模数据时非常耗时）。
    ///
    /// 2.加速压缩过程
    ///
    /// 简单移动仅需更新文件的层级元数据（通过 VersionEdit），无需处理文件内容，大幅缩短压缩时间。
    ///
    /// 3.维持层级平衡
    ///
    /// 条件 3 确保移动操作不会导致祖父层级过度重叠，避免引发连锁的大规模压缩，维持 LSM 树的整体稳定性。
    pub fn is_trivial_move(&self) -> bool {
        // self.manual 标记压缩任务是否为手动触发（如用户调用 compact_range）；
        // 手动压缩通常针对特定范围的优化，需要严格合并数据，因此即使满足条件也不执行简单移动。
        if self.manual {
            return false;  // 手动触发的压缩不适用简单移动优化
        }
        // self.grandparents 存储与当前压缩范围重叠的 “祖父层级”（即 level+2 层级）文件；
        // total_size 计算这些文件的总大小，用于后续判断是否满足移动条件。
        let inputs_size;
        if let Some(gp) = self.grandparents.as_ref() {
            inputs_size = total_size(gp.iter());  // 祖父层级（level+2）相关文件的总大小
        } else {
            inputs_size = 0;   // 无祖父层级文件，总大小为 0
        }
        // 条件 1（num_inputs(0) == 1）：源层级（level）中参与压缩的文件只有 1 个，确保移动对象单一；
        // 条件 2（num_inputs(1) == 0）：目标层级（level+1）中没有与当前文件重叠的文件，无需合并即可移动；
        // 条件 3（inputs_size < 10 * max_file_size）：祖父层级（level+2）的相关文件总大小较小，避免移动后导致祖父层级出现大量重叠文件（否则未来压缩代价会剧增）。
        self.num_inputs(0) == 1 && self.num_inputs(1) == 0 && inputs_size < 10 * self.max_file_size
    }

    /// 这段代码定义了 should_stop_before 方法，用于在 LSM 树的压缩（compaction）过程中判断是否需要在当前键之前停止构建当前输出 SSTable，是控制压缩输出文件拆分的核心逻辑，
    /// 目的是避免生成的新 SSTable 与更高层级（grandparent，通常是 level+2 及以上）的 SSTable 产生过多重叠，从而优化后续压缩性能。
    ///
    /// 这个方法的核心作用是通过控制输出文件的拆分，减少与更高层级 SSTable 的重叠，从而避免未来压缩操作需要处理过多重叠文件，降低系统开销。这是 LSM 树中优化压缩性能的关键策略之一（称为 “避免重叠膨胀”）。
    ///
    /// 简单来说，它是压缩过程中的 “文件拆分控制器”，当检测到当前输出文件与更高层级文件的重叠数据过多时，会触发新文件的创建，以保持存储结构的高效性。
    pub fn should_stop_before(&mut self, k: InternalKey<'_>) -> bool {
        // 如果没有更高层级的参考文件（grandparents.is_none()），则无需拆分，标记已处理键（self.seen_key = true）并返回 false（不停止）。
        if self.grandparents.is_none() {
            self.seen_key = true;
            return false;
        }
        let grandparents = self.grandparents.as_ref().unwrap();
        // 遍历更高层级的 SSTable（grandparents），通过内部键比较器（icmp）找到与当前键（k）相关的重叠文件：
        // 当当前键大于某个更高层级文件的最大键（largest）时，说明该文件与当前压缩范围存在重叠。
        while self.grandparent_ix < grandparents.len() && self.icmp.cmp(k, &grandparents[self.grandparent_ix].borrow().largest) == Ordering::Greater {
            // 若已处理过键（self.seen_key），则累加重叠文件的大小到 overlapped_bytes（记录总重叠数据量）。
            if self.seen_key {
                self.overlapped_bytes += grandparents[self.grandparent_ix].borrow().size;
            }
            // 递增 grandparent_ix 以继续检查下一个更高层级文件。
            self.grandparent_ix += 1;
        }
        // 标记已处理键（self.seen_key = true）。
        self.seen_key = true;
        // 如果总重叠数据量（overlapped_bytes）超过 10 * self.max_file_size（通常是输出文件最大尺寸的 10 倍），则：
        if self.overlapped_bytes > 10 * self.max_file_size {
            // 重置重叠数据量计数器。
            self.overlapped_bytes = 0;
            // 返回 true，表示需要停止当前输出文件的构建，新建一个输出文件（避免过度重叠）。
            true
        } else {
            // 否则返回 false，继续向当前输出文件添加数据。
            false
        }
    }
}

/// VersionSet managed the various versions that are live within a database. A single version
/// contains references to the files on disk as they were at a certain point.
/// VersionSet 结构体是类似 LevelDB/RocksDB 等存储引擎中的核心元数据管理结构，
/// 用于跟踪和管理数据库的所有版本信息、文件编号、日志状态以及压缩相关的指针。它是数据库版本控制和一致性的关键组件。
pub struct VersionSet {
    /// 数据库所在的文件系统路径，用于定位数据文件。
    dbname: PathBuf,
    /// 数据库的配置选项（如块大小、压缩方式等）。
    opt: Options,
    /// 内部键比较器，用于排序和比较键（考虑版本号等元数据）。
    cmp: InternalKeyCmp,
    /// 表缓存，用于缓存打开的 SSTable 文件，提升读取性能。
    cache: Shared<TableCache>,

    /// 下一个待分配的文件编号，确保每个新文件有唯一标识。
    pub next_file_num: u64,
    /// 当前 manifest 文件（数据库元数据日志）的编号。
    pub manifest_num: u64,
    /// 最后一个写入操作的序列号，用于保证操作的顺序性和原子性。
    pub last_seq: u64,
    /// 当前活跃的 WAL（Write-Ahead Log）日志文件编号。
    pub log_num: u64,
    /// 上一个日志文件编号（可能用于恢复或清理）。
    pub prev_log_num: u64,

    /// 当前生效的数据库版本（Version 包含某一时刻的所有 SSTable 文件布局）。
    current: Option<Shared<Version>>,
    /// 每个层级（Level）的压缩指针，标记该层级下次压缩的起始位置。
    compaction_ptrs: [Bytes; NUM_LEVELS],

    /// 用于写入 manifest 日志的 writer，记录数据库版本变更（如文件添加 / 删除）和将元数据记录写入磁盘文件，确保崩溃后可恢复。
    descriptor_log: Option<LogWriter<Box<dyn Write>>>,
}

impl VersionSet {
    // Note: opt.cmp should not contain an InternalKeyCmp at this point, but instead the default or
    // user-supplied one.
    pub fn new<P: AsRef<Path>>(db: P, opt: Options, cache: Shared<TableCache>) -> VersionSet {
        let v = share(Version::new(cache.clone(), opt.cmp.clone()));
        VersionSet {
            dbname: db.as_ref().to_owned(),
            cmp: InternalKeyCmp(opt.cmp.clone()),
            opt,
            cache,

            next_file_num: 2,
            manifest_num: 0,
            last_seq: 0,
            log_num: 0,
            prev_log_num: 0,

            current: Some(v),
            compaction_ptrs: Default::default(),
            descriptor_log: None,
        }
    }

    pub fn current_summary(&self) -> String {
        self.current.as_ref().unwrap().borrow().level_summary()
    }

    /// live_files returns the files that are currently active.
    ///
    /// 返回当前所有活跃的文件
    pub fn live_files(&self) -> HashSet<FileNum> {
        let mut files = HashSet::new();
        if let Some(ref version) = self.current {
            for level in 0..NUM_LEVELS {
                for file in &version.borrow().files[level] {
                    files.insert(file.borrow().num);
                }
            }
        }
        files
    }

    /// current returns a reference to the current version. It panics if there is no current version.
    ///
    /// 这段代码定义了一个名为 current 的公共方法，用于获取当前版本（Version）的共享引用（Shared<Version>）。
    ///
    /// 方法返回类型是 Shared<Version>，这通常是一种线程安全的共享指针类型（可能是 Arc<Rc<Version>> 或类似的引用计数智能指针），允许多个持有者共享对 Version 实例的访问
    pub fn current(&self) -> Shared<Version> {
        // 首先通过 assert!(self.current.is_some()) 进行断言检查，确保 self.current 中确实包含一个版本实例。如果 self.current 为空（None），则会触发程序 panic
        assert!(self.current.is_some());
        // 若断言通过，则通过 self.current.as_ref().unwrap().clone() 获取 self.current 中存储的版本引用并克隆它（这里的克隆是引用计数的增加，而非深拷贝）
        self.current.as_ref().unwrap().clone()
    }

    pub fn add_version(&mut self, v: Version) {
        self.current = Some(share(v));
    }

    /// 这个 new_file_number 方法用于生成一个唯一的文件编号（FileNum），用于标识数据库中创建的新文件（如 SSTable 文件、日志文件等）。
    ///
    /// 这样做的目的是：
    ///
    /// 确保每次调用都能生成一个唯一的编号（通过递增方式）
    ///
    /// 避免文件编号冲突（数据库中每个文件都有唯一编号）
    ///
    /// 维护一个全局递增的编号序列，方便文件管理和版本控制
    pub fn new_file_number(&mut self) -> FileNum {
        self.next_file_num += 1;
        self.next_file_num - 1
    }

    /// 这个 reuse_file_number 方法用于回收并重用文件编号，当某个已分配的文件编号最终没有被使用时（例如创建文件失败或文件为空），可以通过它将编号归还给系统，避免编号资源浪费。
    ///
    /// 作用场景：
    ///
    /// 例如，当创建文件失败（如 build_table 生成了空文件），通过调用该方法可以回收本次分配的编号，让它在后续创建新文件时被重新使用，避免编号无意义地递增（防止编号快速耗尽或过大）。
    pub fn reuse_file_number(&mut self, n: FileNum) {
        // 当参数 n 等于 “上一个分配的文件编号”（即 self.next_file_num - 1）时，说明这个编号是最近刚分配但未被有效使用的。
        if n == self.next_file_num - 1 {
            // 此时将 self.next_file_num 重置为 n，意味着下一次调用 new_file_number 时，会重新分配这个编号。
            self.next_file_num = n;
        }
    }

    pub fn mark_file_number_used(&mut self, n: FileNum) {
        if self.next_file_num <= n {
            self.next_file_num = n + 1;
        }
    }

    /// needs_compaction returns true if a compaction makes sense at this point.
    ///
    /// 这个 needs_compaction 方法用于判断数据库是否需要执行压缩（Compaction）操作，是压缩策略的 “触发器判断条件”。
    ///
    /// 其逻辑非常直接：
    ///
    /// 先获取当前版本（self.current）的压缩相关状态
    ///
    /// 判断两个条件是否满足其一：
    ///
    /// 版本的压缩分数（compaction_score）≥ 1.0（说明某个层级的文件数量或大小超过阈值）
    ///
    /// 存在标记为需要压缩的文件（file_to_compact.is_some()，说明某个文件查询次数过多）
    pub fn needs_compaction(&self) -> bool {
        assert!(self.current.is_some());
        let v = self.current.as_ref().unwrap();
        let v = v.borrow();
        v.compaction_score.unwrap_or(0.0) >= 1.0 || v.file_to_compact.is_some()
    }

    fn approximate_offset(&self, v: &Shared<Version>, key: InternalKey<'_>) -> usize {
        let mut offset = 0;
        for level in 0..NUM_LEVELS {
            for f in &v.borrow().files[level] {
                if self.opt.cmp.cmp(&f.borrow().largest, key) <= Ordering::Equal {
                    offset += f.borrow().size;
                } else if self.opt.cmp.cmp(&f.borrow().smallest, key) == Ordering::Greater {
                    // In higher levels, files are sorted; we don't need to search further.
                    if level > 0 {
                        break;
                    }
                } else if let Ok(tbl) = self.cache.borrow_mut().get_table(f.borrow().num) {
                    offset += tbl.approx_offset_of(key);
                }
            }
        }
        offset
    }
    /// 这个 pick_compaction 方法是数据库压缩机制的核心逻辑，用于选择并创建一个合适的压缩任务（Compaction）。
    /// 它根据当前版本的状态，决定压缩哪些文件、从哪个层级开始压缩，为后续的压缩执行提供具体任务描述。
    pub fn pick_compaction(&mut self) -> Option<Compaction> {
        assert!(self.current.is_some());
        let current = self.current();
        let current = current.borrow();
        // 初始化压缩任务 创建一个新的压缩任务对象 c，并声明变量 level 用于存储需要压缩的层级。
        let mut c = Compaction::new(&self.opt, 0, self.current.clone());
        let level;

        // Size compaction?
        // 当压缩分数 ≥1.0 时（某个层级需要按大小/数量压缩）
        if current.compaction_score.unwrap_or(0.0) >= 1.0 {
            level = current.compaction_level.unwrap(); // 获取需要压缩的层级
            assert!(level < NUM_LEVELS - 1);
            // 从该层级选择第一个超过压缩指针的文件作为压缩输入
            for f in &current.files[level] {
                if self.compaction_ptrs[level].is_empty() || self.cmp.cmp(&f.borrow().largest, &self.compaction_ptrs[level]) == Ordering::Greater {
                    c.add_input(0, f.clone()); // 添加到压缩输入
                    break;
                }
            }
            // 如果没有找到符合条件的文件，默认选择该层级的第一个文件
            if c.num_inputs(0) == 0 {
                // Add first file in level. This will also reset the compaction pointers.
                c.add_input(0, current.files[level][0].clone());
            }
        } else if let Some(ref ftc) = current.file_to_compact { // 当存在标记为需要压缩的文件时（因查询次数过多）
            // Seek compaction?
            level = current.file_to_compact_lvl;  // 获取文件所在层级
            c.add_input(0, ftc.clone()); // 将该文件作为压缩输入
        } else {
            return None;
        }

        c.level = level;
        c.input_version.clone_from(&self.current);

        if level == 0 {
            let (smallest, largest) = get_range(&self.cmp, c.inputs[0].iter());
            // This call intentionally overwrites the file previously put into c.inputs[0].
            // 选择 Level 0 中与输入文件键范围重叠的所有文件（因 Level 0 允许文件重叠）
            c.inputs[0] = current.overlapping_inputs(0, &smallest, &largest);
            assert!(!c.inputs[0].is_empty());
        }

        self.setup_other_inputs(&mut c);  // 设置其他层级的关联输入文件（如更高层级中与输入键范围重叠的文件）
        Some(c)
    }

    /// 这个 compact_range 方法是 LSM 树版本管理器（VersionSet）中用于创建指定层级和键范围内的压缩任务（Compaction） 的核心函数。
    /// 它根据输入的层级、起始键和结束键，筛选需要压缩的 SSTable 文件，生成压缩任务，为后续的实际压缩操作（合并文件、清理无效数据、下沉到下一层级）提供基础。
    ///
    /// compact_range 的核心价值是生成指定范围的压缩任务，主要作用包括：
    ///
    /// 精准筛选压缩文件
    ///
    /// 只选择与目标键范围 [from, to] 重叠的文件，避免无意义的全量压缩，减少资源消耗。
    ///
    /// 控制压缩规模
    ///
    /// 对 Level 1+ 限制压缩文件总大小，确保单次压缩操作的代价可控，避免影响数据库正常服务。
    ///
    /// 构建完整压缩上下文
    ///
    /// 通过 setup_other_inputs 关联下一层级的重叠文件，确保压缩后的数据能与下一层级正确合并，维持 LSM 树的层级结构特性（高层级文件范围不重叠）。
    ///
    /// 支持手动压缩触发
    ///
    /// 作为 compact_range 手动压缩接口的底层实现，允许用户针对特定范围主动触发压缩，优化热点数据的存储结构。
    pub fn compact_range(
        &mut self,
        level: usize, // 目标压缩层级
        from: InternalKey<'_>, // 压缩范围的起始内部键
        to: InternalKey<'_>, // 压缩范围的结束内部键
    ) -> Option<Compaction> {
        assert!(self.current.is_some());  // 确保存在当前数据库版本
        // self.current 是数据库当前版本（Version）的引用，存储了各层级的 SSTable 元数据；
        // overlapping_inputs 方法筛选出层级 level 中所有键范围与 [from, to] 重叠的 SSTable 文件（inputs）；
        // 若没有重叠文件，则无需压缩，直接返回 None。
        // 从当前版本中获取与 [from, to] 范围重叠的 SSTable 文件
        let mut inputs = self
            .current
            .as_ref()
            .unwrap()
            .borrow()
            .overlapping_inputs(level, from, to);
        if inputs.is_empty() {
            return None;  // 无重叠文件，无需压缩，返回 None
        }
        // Level 1+ 特性：这些层级的 SSTable 通常较大且数量多，为避免单次压缩操作过于繁重（消耗过多 I/O 和 CPU），需要限制参与压缩的文件总大小；
        // 限制逻辑：累加文件大小，当超过配置的 max_file_size 时，截断文件列表，只保留前 i+1 个文件，确保压缩任务的资源消耗可控；
        // Level 0 不限制：Level 0 的文件通常较小且数量少，因此不做此限制。
        if level > 0 { // Level 1 及以上层级需要限制单次压缩的文件总大小
            let mut total = 0;
            for i in 0..inputs.len() {
                total += inputs[i].borrow().size; // 累加文件大小
                // 若总大小超过配置的最大文件大小，截断文件列表
                if total > self.opt.max_file_size {
                    inputs.truncate(i + 1);
                    break;
                }
            }
        }
        // 压缩任务初始化：Compaction::new 创建一个新的压缩任务，包含配置（opt）、目标层级（level）和当前版本信息；
        // 设置输入文件：c.inputs[0] 存储当前层级中需要被压缩的文件（inputs）；
        // 标记手动压缩：c.manual = true 区分手动触发（如 compact_range 调用）和自动触发（如文件数量超限）的压缩，可能影响压缩策略；
        // 补充其他输入：setup_other_inputs 方法会查找下一层级（level+1）中与当前压缩范围重叠的文件，将其也加入压缩任务（因为压缩后的数据需要与下一层级的重叠文件合并，避免后续再次重叠）。

        // 创建新的压缩任务，关联配置、目标层级和当前版本
        let mut c = Compaction::new(&self.opt, level, self.current.clone());
        c.inputs[0] = inputs;  // 设置当前层级的输入文件（inputs[0] 表示待压缩的源层级文件）
        c.manual = true; // 标记为手动触发的压缩（区别于自动后台压缩）
        self.setup_other_inputs(&mut c); // 设置其他相关层级的输入文件（如下一层级的重叠文件）
        Some(c)
    }

    /// 这个 setup_other_inputs 方法是 LSM 树版本管理器（VersionSet）中用于完善压缩任务（Compaction）上下文信息的核心函数。
    /// 它会补充目标层级的重叠文件、尝试扩展当前层级的压缩范围以优化效率，并关联祖父层级的相关文件，最终为压缩任务构建完整的执行边界和依赖关系。
    fn setup_other_inputs(&mut self, compaction: &mut Compaction) {
        assert!(self.current.is_some());
        let current = self.current.as_ref().unwrap();
        let current = current.borrow(); // 获取当前数据库版本

        let level = compaction.level;  // 压缩任务的源层级（待压缩的层级）
        // 1. 获取源层级输入文件的键范围 [smallest, largest]
        let (mut smallest, mut largest) = get_range(&self.cmp, compaction.inputs[0].iter());

        // Set up level+1 inputs.
        // 2. 补充目标层级（level+1）的输入文件：筛选与源层级范围重叠的文件
        compaction.inputs[1] = current.overlapping_inputs(level + 1, &smallest, &largest);
        // 2.1 计算源层级+目标层级的整体键范围 [allstart, alllimit]
        let (mut allstart, mut alllimit) = get_range(
            &self.cmp,
            compaction.inputs[0]
                .iter()
                .chain(compaction.inputs[1].iter()),
        );

        // Check if we can add more inputs in the current level without having to compact more
        // inputs from level+1.
        // 2.2 检查是否可扩展源层级的输入文件
        if !compaction.inputs[1].is_empty() {
            // 找到源层级中与 [allstart, alllimit] 重叠的所有文件（可能比原 inputs[0] 更多）
            let expanded0 = current.overlapping_inputs(level, &allstart, &alllimit);
            // 计算目标层级输入文件大小 + 扩展后源层级文件大小
            let inputs1_size = total_size(compaction.inputs[1].iter());
            let expanded0_size = total_size(expanded0.iter());
            // ...if we picked up more files in the current level, and the total size is acceptable
            // 扩展条件：
            // - 扩展后源文件数量增加（确实能合并更多文件）
            // - 总大小 < 25 * 最大文件大小（避免压缩规模过大）
            // - 目标层级重叠文件数量不变（扩展后无需额外合并目标层级文件）
            if expanded0.len() > compaction.num_inputs(0) && (inputs1_size + expanded0_size) < 25 * self.opt.max_file_size {
                let (new_start, new_limit) = get_range(&self.cmp, expanded0.iter());
                let expanded1 = current.overlapping_inputs(level + 1, &new_start, &new_limit);
                if expanded1.len() == compaction.num_inputs(1) {
                    log!(
                        self.opt.log,
                        "Expanding inputs@{} {}+{} ({}+{} bytes) to {}+{} ({}+{} bytes)",
                        level,
                        compaction.inputs[0].len(),
                        compaction.inputs[1].len(),
                        total_size(compaction.inputs[0].iter()),
                        total_size(compaction.inputs[1].iter()),
                        expanded0.len(),
                        expanded1.len(),
                        total_size(expanded0.iter()),
                        total_size(expanded1.iter())
                    );
                    // 满足条件：更新压缩范围和输入文件
                    smallest = new_start;
                    largest = new_limit;
                    compaction.inputs[0] = expanded0;
                    compaction.inputs[1] = expanded1;
                    // 更新整体范围
                    let (newallstart, newalllimit) = get_range(
                        &self.cmp,
                        compaction.inputs[0]
                            .iter()
                            .chain(compaction.inputs[1].iter()),
                    );
                    allstart = newallstart;
                    alllimit = newalllimit;
                }
            }
        }

        // Set the list of grandparent (l+2) inputs to the files overlapped by the current overall range.
        // 若祖父层级（level+2）存在，筛选与整体范围 [allstart, alllimit] 重叠的文件
        if level + 2 < NUM_LEVELS {
            // grandparents 作用：用于判断压缩任务是否可执行 “简单移动”（is_trivial_move）优化 —— 若祖父层级重叠文件总大小较小，且其他条件满足，可直接将源文件移动到目标层级，无需合并。
            let grandparents = self.current.as_ref().unwrap().borrow().overlapping_inputs(
                level + 2,
                &allstart,
                &alllimit,
            );
            compaction.grandparents = Some(grandparents);   // 存储到压缩任务中
        }
        // 记录压缩任务的核心信息到日志
        log!(self.opt.log,"Compacting @{} {:?} .. {:?}",level,smallest,largest);
        // 1. 记录当前层级的压缩指针：标记压缩已处理到 `largest` 键（下次压缩从这里开始）
        compaction.edit().set_compact_pointer(level, &largest);
        // 2. 更新内存中的压缩指针（快速访问）
        self.compaction_ptrs[level] = largest;
    }

    /// write_snapshot writes the current version, with all files, to the manifest.
    /// 这个 write_snapshot 方法用于将数据库当前的版本状态快照写入到 manifest 日志文件（元数据日志），
    /// 确保数据库的版本信息能够持久化存储，在崩溃后可以恢复。它的核心作用是记录数据库当前的完整元数据状态。
    fn write_snapshot(&mut self) -> Result<usize> {
        assert!(self.descriptor_log.is_some());

        let mut edit = VersionEdit::new();
        edit.set_comparator_name(self.opt.cmp.id()); // 记录比较器名称（确保兼容性）

        // Save compaction pointers.
        // 压缩指针记录了每个层级（Level）下次压缩的起始位置，是数据库压缩策略的重要状态。
        // 遍历所有层级，将非空的压缩指针记录到 VersionEdit 中。
        for level in 0..NUM_LEVELS {
            if !self.compaction_ptrs[level].is_empty() {
                edit.set_compact_pointer(level, &self.compaction_ptrs[level]);
            }
        }

        let current = self.current.as_ref().unwrap().borrow(); // 获取当前版本
        // Save files.
        // 这段代码遍历所有层级的所有文件，将每个文件的完整元数据（如文件编号、大小、最小键、最大键等）记录到 VersionEdit 中，从而完整保存当前版本的文件布局。
        for level in 0..NUM_LEVELS {
            let fs = &current.files[level]; // 当前层级的所有文件
            for f in fs {
                // 将每个文件的元数据（编号、大小、键范围等）添加到 VersionEdit
                edit.add_file(level, f.borrow().clone());
            }
        }
        // 先通过 edit.encode() 将 VersionEdit 序列化为字节流，再通过 add_record 写入到 manifest 日志中。
        self.descriptor_log
            .as_mut()
            .unwrap()
            .add_record(&edit.encode())
    }

    /// log_and_apply merges the given edit with the current state and generates a new version. It
    /// writes the VersionEdit to the manifest.
    ///
    /// log_and_apply 方法是数据库版本管理的核心操作，主要做以下几件事：
    ///
    /// 准备版本变更信息：完善 VersionEdit（版本变更指令），补充日志编号、序列号序列号等必要信息。
    ///
    /// 生成新版本：基于当前版本和变更指令，通过 Builder 构建新的数据库版本（Version），并计算其压缩策略。
    ///
    /// 持久化元数据：将版本变更记录录到 manifest 日志文件，确保变更可恢复；更新 CURRENT 文件指向最新 manifest。
    ///
    /// 应用新版本：将新生成的版本添加到版本集合，更新数据库当前的日志编号，完成版本切换。
    pub fn log_and_apply(&mut self, mut edit: VersionEdit) -> Result<()> {
        assert!(self.current.is_some());

        if edit.log_number.is_none() {
            edit.set_log_num(self.log_num);
        } else {
            assert!(edit.log_number.unwrap() >= self.log_num);
            assert!(edit.log_number.unwrap() < self.next_file_num);
        }
        if edit.prev_log_number.is_none() {
            edit.set_prev_log_num(self.prev_log_num);
        }
        edit.set_next_file(self.next_file_num);
        edit.set_last_seq(self.last_seq);

        let mut v = Version::new(self.cache.clone(), self.opt.cmp.clone());
        {
            let mut builder = Builder::new();
            // 整合版本变更信息（包括压缩指针、文件删除和新增操作），为后续生成新 Version 做准备。
            builder.apply(&edit, &mut self.compaction_ptrs);
            // 整合新旧文件，构建一个逻辑一致的新版本数据布局。
            builder.save_to(&self.cmp, self.current.as_ref().unwrap(), &mut v);
        }
        // 计算数据库当前版本（Version）中需要进行压缩（Compaction）的层级
        self.finalize(&mut v);

        if self.descriptor_log.is_none() {
            // 设置源文件名称
            let descname = manifest_file_name(&self.dbname, self.manifest_num);
            edit.set_next_file(self.next_file_num);
            self.descriptor_log = Some(LogWriter::new(
                // 打开源文件
                self.opt.env.open_writable_file(Path::new(&descname))?,
            ));
            // 将数据库当前的版本状态快照写入到 manifest 日志文件（元数据日志）， 确保数据库的版本信息能够持久化存储，在崩溃后可以恢复。
            self.write_snapshot()?;
        }

        let encoded = edit.encode();
        if let Some(ref mut lw) = self.descriptor_log {
            // 添加数据到日志文件
            lw.add_record(&encoded)?;
            // 刷新文件流数据到物理磁盘
            lw.flush()?;
        }
        // 设置文件当前文件里的内容为 最新的 manifest 文件名
        set_current_file(
            self.opt.env.as_ref().as_ref(),
            &self.dbname,
            self.manifest_num,
        )?;

        self.add_version(v);
        // log_number was set above.
        self.log_num = edit.log_number.unwrap();

        // TODO: Roll back written files if something went wrong.
        Ok(())
    }

    /// 这个 finalize 方法用于计算数据库当前版本（Version）中需要进行压缩（Compaction）的层级，是存储引擎压缩策略的核心逻辑之一。
    /// 它通过计算每个层级的 “压缩分数”，确定哪个层级最需要进行压缩，以优化查询性能和存储空间。
    fn finalize(&self, v: &mut Version) {
        let mut best_lvl = None; // 最终确定的需要压缩的层级
        let mut best_score = None; // 对应的最高压缩分数

        for l in 0..NUM_LEVELS - 1 { // 遍历除最后一层外的所有层级
            let score = if l == 0 {
                // Level 0 的分数计算：文件数量 / 4.0
                v.files[l].len() as f64 / 4.0
            } else {
                // Level 1+ 的分数计算：当前层级总大小 / 该层级允许的最大大小
                let mut max_bytes = 10.0 * f64::from(1 << 20); // 初始值 10MB（10 * 1MB）
                for _ in 0..l - 1 {
                    max_bytes *= 10.0; // 每高一层，最大允许大小乘以 10（Level 1:10MB, Level 2:100MB...）
                }
                total_size(v.files[l].iter()) as f64 / max_bytes // 总大小 / 最大允许大小
            };
            if let Some(ref mut b) = best_score {
                // 如果当前分数高于已记录的最高分数，则更新
                if *b < score {
                    *b = score;
                    best_lvl = Some(l);
                }
            } else {
                // 初始化最高分数和层级
                best_score = Some(score);
                best_lvl = Some(l);
            }
        }
        v.compaction_score = best_score; // 记录最高压缩分数
        v.compaction_level = best_lvl; // 记录最需要压缩的层级
    }

    /// recover recovers the state of a LevelDB instance from the files on disk. If recover()
    /// returns true, the a manifest needs to be written eventually (using log_and_apply()).
    pub fn recover(&mut self) -> Result<bool> {
        assert!(self.current.is_some());

        let mut current = read_current_file(self.opt.env.as_ref().as_ref(), &self.dbname)?;
        let len = current.len();
        current.truncate(len - 1);
        let current = Path::new(&current);

        let descfilename = self.dbname.join(current);
        let mut builder = Builder::new();
        {
            let mut descfile = self
                .opt
                .env
                .open_sequential_file(Path::new(&descfilename))?;
            let mut logreader = LogReader::new(
                &mut descfile,
                // checksum=
                true,
            );

            let mut log_number = None;
            let mut prev_log_number = None;
            let mut next_file_number = None;
            let mut last_seq = None;

            let mut buf = Vec::new();
            while let Ok(size) = logreader.read(&mut buf) {
                if size == 0 {
                    break;
                }
                let edit = VersionEdit::decode_from(&buf)?;
                builder.apply(&edit, &mut self.compaction_ptrs);
                if let Some(ln) = edit.log_number {
                    log_number = Some(ln);
                }
                if let Some(nfn) = edit.next_file_number {
                    next_file_number = Some(nfn);
                }
                if let Some(ls) = edit.last_seq {
                    last_seq = Some(ls);
                }
                if let Some(pln) = edit.prev_log_number {
                    prev_log_number = Some(pln);
                }
            }

            if let Some(ln) = log_number {
                self.log_num = ln;
                self.mark_file_number_used(ln);
            } else {
                return err(
                    StatusCode::Corruption,
                    "no meta-lognumber entry in descriptor",
                );
            }
            if let Some(nfn) = next_file_number {
                self.next_file_num = nfn + 1;
            } else {
                return err(
                    StatusCode::Corruption,
                    "no meta-next-file entry in descriptor",
                );
            }
            if let Some(ls) = last_seq {
                self.last_seq = ls;
            } else {
                return err(
                    StatusCode::Corruption,
                    "no last-sequence entry in descriptor",
                );
            }
            if let Some(pln) = prev_log_number {
                self.prev_log_num = pln;
                self.mark_file_number_used(prev_log_number.unwrap());
            } else {
                self.prev_log_num = 0;
            }
        }

        let mut v = Version::new(self.cache.clone(), self.opt.cmp.clone());
        builder.save_to(&self.cmp, self.current.as_ref().unwrap(), &mut v);
        self.finalize(&mut v);
        self.add_version(v);
        self.manifest_num = self.next_file_num - 1;
        log!(
            self.opt.log,
            "Recovered manifest with next_file={} manifest_num={} log_num={} prev_log_num={} \
             last_seq={}",
            self.next_file_num,
            self.manifest_num,
            self.log_num,
            self.prev_log_num,
            self.last_seq
        );

        // A new manifest needs to be written only if we don't reuse the existing one.
        Ok(!self.reuse_manifest(&descfilename, current))
    }

    /// reuse_manifest checks whether the current manifest can be reused.
    fn reuse_manifest(
        &mut self,
        current_manifest_path: &Path,
        current_manifest_base: &Path,
    ) -> bool {
        // Note: The original has only one option, reuse_logs; reuse_logs has to be set in order to
        // reuse manifests.
        // However, there's not much that stops us from reusing manifests without reusing logs or
        // vice versa. One issue exists though: If no write operations are done, empty log files
        // will accumulate every time a DB is opened, until at least one write happens (otherwise,
        // the logs won't be compacted and deleted).
        if !self.opt.reuse_manifest {
            return false;
        }
        // The original doesn't reuse manifests; we do.
        if let Ok((num, typ)) = parse_file_name(current_manifest_base) {
            if typ != FileType::Descriptor {
                return false;
            }
            if let Ok(size) = self.opt.env.size_of(Path::new(current_manifest_path)) {
                if size >= self.opt.max_file_size {
                    return false;
                }

                assert!(self.descriptor_log.is_none());
                let s = self
                    .opt
                    .env
                    .open_appendable_file(Path::new(current_manifest_path));
                if let Ok(f) = s {
                    log!(self.opt.log, "reusing manifest {:?}", current_manifest_path);
                    self.descriptor_log = Some(LogWriter::new_with_off(f, size));
                    self.manifest_num = num;
                    return true;
                } else {
                    log!(self.opt.log, "reuse_manifest: {}", s.err().unwrap());
                }
            }
        }
        false
    }

    /// make_input_iterator returns an iterator over the inputs of a compaction.
    ///
    /// 这段代码定义了 make_input_iterator 方法，用于为 LSM 树存储引擎中的压缩操作（compaction）创建一个统一的输入迭代器，以便合并多个 SSTable 中的数据。
    pub fn make_input_iterator(&self, c: &Compaction) -> Box<dyn LdbIterator> {
        // 根据压缩任务（Compaction）的级别（c.level）确定迭代器容器的初始容量（cap）：
        // 若处理的是 L0 层（Level 0），容量为输入文件数 + 1（L0 层文件可能重叠，需要单独处理）
        // 其他层级容量为 2（通常是当前层与下一层的文件合并）
        let cap = if c.level == 0 { c.num_inputs(0) + 1 } else { 2 };
        let mut iters: Vec<Box<dyn LdbIterator>> = Vec::with_capacity(cap);
        // 遍历压缩任务的两个输入源（i in 0..2，通常对应两个不同层级的 SSTable 集合）
        // 对每个输入源：
        for i in 0..2 {
            if c.num_inputs(i) == 0 {
                continue;
            }
            // 若为 L0 层（c.level + i == 0）：由于 L0 层的 SSTable 键范围可能重叠，需要为每个文件创建单独的迭代器并添加到容器中
            if c.level + i == 0 {
                // Add individual iterators for L0 tables.
                for fi in 0..c.num_inputs(i) {
                    let f = &c.inputs[i][fi];
                    let s = self.cache.borrow_mut().get_table(f.borrow().num);
                    if let Ok(tbl) = s {
                        iters.push(Box::new(tbl.iter()));
                    } else {
                        log!(self.opt.log,"error opening table {}: {}",f.borrow().num,s.err().unwrap());
                    }
                }
            } else {
                // Create concatenating iterator higher levels.
                // 其他层级：创建一个串联迭代器（new_version_iter），统一处理该层级的所有 SSTable（这些文件的键范围通常不重叠且有序）
                iters.push(Box::new(new_version_iter(
                    c.inputs[i].clone(),
                    self.cache.clone(),
                    self.opt.cmp.clone(),
                )));
            }
        }
        assert!(iters.len() <= cap);
        // 使用 MergingIter::new 将所有创建的迭代器合并为一个单一的迭代器
        // 传入比较器（cmp）确保合并后的数据按键的顺序排列
        let cmp: Rc<Box<dyn Cmp>> = Rc::new(Box::new(self.cmp.clone()));
        // 返回封装了所有输入数据的合并迭代器，作为压缩操作的统一数据输入源
        Box::new(MergingIter::new(cmp, iters))
    }
}

/// 这个 Builder 结构体用于构建或修改数据库的版本信息，主要追踪每个层级（Level）中需要添加和删除的文件，
/// 常见于类似 LevelDB/RocksDB 的存储引擎中，用于管理 SSTable 文件的版本变更。
struct Builder {
    // (added, deleted) files per level.
    /// 一个数组，元素是存储文件编号（FileNum）的向量，长度为 NUM_LEVELS（数据库的总层级数）。
    /// 它记录了每个层级中需要删除的文件：例如，在压缩（Compaction）过程中，旧的 SSTable 文件被合并后，其编号会被加入到对应层级的 deleted 列表中，等待从版本中移除。
    deleted: [Vec<FileNum>; NUM_LEVELS],
    /// 同样是长度为 NUM_LEVELS 的数组，元素是存储文件元数据句柄（FileMetaHandle）的向量。
    /// 它记录了每个层级中需要新增的文件：例如，压缩后生成的新 SSTable 文件，其元数据（如编号、大小、键范围等）会通过 FileMetaHandle 包装，加入到对应层级的 added 列表中，等待添加到新版本中。
    added: [Vec<FileMetaHandle>; NUM_LEVELS],
}

impl Builder {
    fn new() -> Builder {
        Builder {
            deleted: Default::default(),
            added: Default::default(),
        }
    }

    /// apply applies the edits recorded in edit to the builder state. compaction pointers are
    /// copied to the supplied compaction_ptrs array.
    /// 这段代码是 Builder 结构体的 apply 方法，用于将 VersionEdit（数据库版本变更指令）中的修改应用到当前构建器中，最终生成新的数据库版本。
    /// 它的核心作用是整合版本变更信息（包括压缩指针、文件删除和新增操作），为后续生成新 Version 做准备。
    fn apply(&mut self, edit: &VersionEdit, compaction_ptrs: &mut [Bytes; NUM_LEVELS]) {
        // 将这些指针更新到 compaction_ptrs 数组中，用于标记各层级下次压缩的起始位置。
        for c in edit.compaction_ptrs.iter() {
            compaction_ptrs[c.level].clone_from(&c.key);
        }
        // 将这些文件编号添加到 Builder 的 deleted 数组对应层级中，标记这些文件将在新版本中被移除。
        for &(level, num) in edit.deleted.iter() {
            self.deleted[level].push(num);
        }
        for &(level, ref f) in edit.new_files.iter() {
            // 克隆文件元数据，并计算允许的最大 seek 次数（用于判断是否需要压缩）
            let mut f = f.clone();
            f.allowed_seeks = f.size / 16384; // 16384 = 16KB，根据文件大小计算
            if f.allowed_seeks < 100 {
                f.allowed_seeks = 100;  // 最小允许 100 次 seek
            }
            // Remove this file from the list of deleted files.
            // 从删除列表中移除当前新增文件（如果它曾被标记为删除）
            self.deleted[level] = self.deleted[level]
                .iter()
                .filter_map(|d| if *d != f.num { Some(*d) } else { None })
                .collect();
            // 将新增文件添加到对应层级的 added 列表中
            self.added[level].push(share(f));
        }
    }

    /// maybe_add_file adds a file f at level to version v, if it's not already marked as deleted
    /// in this edit. It also asserts that the ordering of files is preserved.
    /// 这个 maybe_add_file 方法用于判断一个文件是否应该被添加到新版本（Version）的指定层级中，它会进行一系列校验以确保数据一致性。
    fn maybe_add_file(
        &mut self,
        cmp: &InternalKeyCmp,
        v: &mut Version,
        level: usize,
        f: FileMetaHandle,
    ) {
        // Only add file if it's not already deleted.
        // 检查文件是否被标记为删除
        if self.deleted[level].iter().any(|d| *d == f.borrow().num) {
            return;
        }
        // 这段代码中的大括号是显式作用域分隔符，主要目的是控制变量 files 的生命周期，避免借用冲突。
        // 具体来说：
        // let files = &v.files[level]; 对 v.files[level] 进行了不可变借用（&）。
        // 如果没有这个大括号，files 的借用会持续到整个 maybe_add_file 函数结束。
        // 而后续代码 v.files[level].push(f); 需要对 v.files[level] 进行可变借用（因为要修改它）
        {
            let files = &v.files[level];
            // 对于 Level 1 及以上层级（不包括 Level 0），如果该层级已有文件：
            // 取出该层级最后一个文件的最大键（largest）
            // 与当前要添加文件的最小键（smallest）进行比较
            // 通过断言确保 "最后一个文件的最大键 < 当前文件的最小键"
            if level > 0 && !files.is_empty() {
                // File must be after last file in level.
                assert_eq!(
                    cmp.cmp(
                        &files[files.len() - 1].borrow().largest,
                        &f.borrow().smallest
                    ),
                    Ordering::Less
                );
            }
        }
        v.files[level].push(f);
    }

    /// save_to saves the edits applied to the builder to v, adding all non-deleted files from
    /// Version base to v.
    /// 这段代码是 Builder 结构体的 save_to 方法，用于基于基础版本（base） 和当前构建器中的变更（added/deleted） 生成一个新的数据库版本（Version）。
    /// 它的核心作用是整合新旧文件，构建一个逻辑一致的新版本数据布局。
    fn save_to(&mut self, cmp: &InternalKeyCmp, base: &Shared<Version>, v: &mut Version) {
        for level in 0..NUM_LEVELS {
            // 按文件的最小键（smallest）排序新增文件
            // 数据库的 SSTable 文件按层级（Level 0 到 Level N）组织，每层文件需要按键范围有序排列。
            // 排序的依据是文件的 smallest 字段（文件包含的最小键），确保后续合并操作能正确维护顺序。
            sort_files_by_smallest(cmp, &mut self.added[level]);
            // The base version should already have sorted files.
            // 确保基础版本的文件已按最小键排序（基础版本本身应保持有序）
            sort_files_by_smallest(cmp, &mut base.borrow_mut().files[level]);

            // 克隆新增文件和基础版本中的文件（避免直接修改原数据）
            let added = self.added[level].clone();
            let basefiles = base.borrow().files[level].clone();
            // 为新版本的文件列表预留足够空间（基础文件数 + 新增文件数）
            v.files[level].reserve(basefiles.len() + self.added[level].len());

            // 将新增文件迭代器和基础文件迭代器合并
            let iadded = added.into_iter();
            let ibasefiles = basefiles.into_iter();
            // 合并逻辑：使用 merge_iters 函数将基础版本的文件和新增文件按 smallest 键升序合并（类似归并排序），确保合并后的文件列表有序。
            let merged = merge_iters(iadded, ibasefiles, |a, b| {
                // 比较两个文件的最小键，按升序合并
                cmp.cmp(&a.borrow().smallest, &b.borrow().smallest)
            });
            // 过滤无效文件：maybe_add_file 方法会检查文件是否应被包含在新版本中（例如，排除已被标记为删除的文件），最终将有效文件添加到新版本的 files[level] 中。
            for m in merged {
                self.maybe_add_file(cmp, v, level, m);
            }

            // Make sure that there is no overlap in higher levels.
            // 跳过 Level 0（允许文件重叠），验证 Level 1+ 的文件无键范围重叠
            if level == 0 {
                continue;
            }
            for i in 1..v.files[level].len() {
                let (prev_end, this_begin) = (
                    &v.files[level][i - 1].borrow().largest, // 前一个文件的最大键
                    &v.files[level][i].borrow().smallest,  // 当前文件的最小键
                );
                // 断言：前一个文件的最大键 < 当前文件的最小键（确保无重叠）
                assert!(cmp.cmp(prev_end, this_begin) < Ordering::Equal);
            }
        }
    }
}

fn manifest_name(file_num: FileNum) -> PathBuf {
    Path::new(&format!("MANIFEST-{:06}", file_num)).to_owned()
}

pub fn manifest_file_name<P: AsRef<Path>>(dbname: P, file_num: FileNum) -> PathBuf {
    dbname.as_ref().join(manifest_name(file_num))
}

fn temp_file_name<P: AsRef<Path>>(dbname: P, file_num: FileNum) -> PathBuf {
    dbname.as_ref().join(format!("{:06}.dbtmp", file_num))
}

fn current_file_name<P: AsRef<Path>>(dbname: P) -> PathBuf {
    dbname.as_ref().join("CURRENT").to_owned()
}

pub fn read_current_file(env: &dyn Env, dbname: &Path) -> Result<String> {
    let mut current = String::new();
    let mut f = env.open_sequential_file(Path::new(&current_file_name(dbname)))?;
    f.read_to_string(&mut current)?;
    if current.is_empty() || !current.ends_with('\n') {
        return err(
            StatusCode::Corruption,
            "current file is empty or has no newline",
        );
    }
    Ok(current)
}

/// 这个 set_current_file 方法是数据库元数据管理的关键函数，用于更新数据库的 “当前 manifest 文件” 指针，确保数据库启动时能找到最新的元数据日志。
/// 它的核心作用是安全地更新标识最新元数据文件的指针，避免因操作中断导致的元数据不一致。
pub fn set_current_file<P: AsRef<Path>>(
    env: &dyn Env,
    dbname: P,
    manifest_file_num: FileNum,
) -> Result<()> {
    let dbname = dbname.as_ref();
    let manifest_base = manifest_name(manifest_file_num);  // 生成 manifest 文件名（如 "MANIFEST-000001"）
    let tempfile = temp_file_name(dbname, manifest_file_num); // 生成临时文件路径
    {
        // 创建并打开临时文件，向其中写入最新的 manifest 文件名（例如将 "MANIFEST-000001" 写入临时文件）。
        // 这一步的作用是：先在临时位置记录 “哪个 manifest 文件是最新的”。
        let mut f = env.open_writable_file(Path::new(&tempfile))?;
        // 写入 manifest 文件名（如 "MANIFEST-000001"）
        f.write_all(manifest_base.display().to_string().as_bytes())?;
        f.write_all(b"\n")?;
    }
    let currentfile = current_file_name(dbname); // 目标文件路径（通常是 "CURRENT"）
    // 将临时文件原子性地重命名为 CURRENT
    if let Err(e) = env.rename(Path::new(&tempfile), Path::new(&currentfile)) {
        // ignore error.
        // 若重命名失败，删除临时文件并返回错误
        let _ = env.delete(Path::new(&tempfile));
        return Err(e);
    }
    Ok(())
}

/// sort_files_by_smallest sorts the list of files by the smallest keys of the files.
fn sort_files_by_smallest<C: Cmp>(cmp: &C, files: &mut [FileMetaHandle]) {
    files.sort_by(|a, b| cmp.cmp(&a.borrow().smallest, &b.borrow().smallest))
}

/// merge_iters merges and collects the items from two sorted iterators.
fn merge_iters<
    Item,
    C: Fn(&Item, &Item) -> Ordering,
    I: Iterator<Item = Item>,
    J: Iterator<Item = Item>,
>(
    mut iter_a: I,
    mut iter_b: J,
    cmp: C,
) -> Vec<Item> {
    let mut a = iter_a.next();
    let mut b = iter_b.next();
    let mut out = vec![];
    while a.is_some() && b.is_some() {
        let ord = cmp(a.as_ref().unwrap(), b.as_ref().unwrap());
        if ord == Ordering::Less {
            out.push(a.unwrap());
            a = iter_a.next();
        } else {
            out.push(b.unwrap());
            b = iter_b.next();
        }
    }
    // 因为迭代器的工作方式是每次调用 next() 才会产生下一个元素。在循环中：
    // a 和 b 是通过 iter_a.next() 和 iter_b.next() 提前获取的 “当前元素”。
    // 当循环结束时，a 或 b 中可能缓存了一个未被处理的元素（例如，iter_a 的下一个元素已经取出并存在 a 中，但 iter_b 已经空了，导致循环退出）。
    // Push cached elements.
    if let Some(a_) = a {
        out.push(a_);
    }
    if let Some(b_) = b {
        out.push(b_);
    }

    // Push remaining elements from either iterator.
    for a in iter_a {
        out.push(a);
    }
    for b in iter_b {
        out.push(b);
    }
    out
}

/// get_range returns the indices of the files within files that have the smallest lower bound
/// respectively the largest upper bound.
///
/// 这个 get_range 方法用于从一组 SSTable 文件中，计算出这组文件整体覆盖的键范围，即所有文件中最小的 “最小键”（smallest）和最大的 “最大键”（largest），最终返回一个包含这两个边界的元组 (最小值Bytes, 最大值Bytes)。
///
/// 示例说明
///
/// 假设输入 3 个 SSTable 文件，每个文件的键范围如下：
///
/// 文件 A：[a, c]（smallest = a，largest = c）
///
/// 文件 B：[b, d]（smallest = b，largest = d）
///
/// 文件 C：[e, f]（smallest = e，largest = f）
///
/// 调用 get_range(cmp, [A, B, C].iter()) 后：
///
/// smallest 取所有文件 smallest 的最小值 → a；
///
/// largest 取所有文件 largest 的最大值 → f；
///
/// 返回结果：(a, f)，即这 3 个文件整体覆盖的键范围是 [a, f]。
fn get_range<'a, C: Cmp, I: Iterator<Item = &'a FileMetaHandle>>(
    c: &C,
    files: I,
) -> (Bytes, Bytes) {
    let mut smallest = None;  // 存储所有文件中的最小 "smallest" 键
    let mut largest = None;  // 存储所有文件中的最大 "largest" 键
    // 边界初始化：首次遍历文件时，将 smallest 和 largest 初始化为第一个文件的最小键和最大键；
    // 边界更新逻辑：
    // 对每个文件的 smallest（文件自身的最小键），与当前记录的 smallest 比较，若更小则更新；
    // 对每个文件的 largest（文件自身的最大键），与当前记录的 largest 比较，若更大则更新；
    // 比较器依赖：所有键的比较均通过传入的 c: &C（实现 Cmp trait 的比较器）执行，确保符合数据库全局的键排序规则（如字典序、数值序等）。
    for f in files {  // 遍历输入的所有 SSTable 文件
        // 第一步：若边界未初始化，用当前文件的键初始化边界
        if smallest.is_none() {
            smallest = Some(f.borrow().smallest.clone());  // 用当前文件的最小键初始化
        }
        if largest.is_none() {
            largest = Some(f.borrow().largest.clone()); // 用当前文件的最大键初始化
        }
        // 第二步：获取当前文件的元数据（含最小/最大键）
        let f = f.borrow();
        // 第三步：更新 "smallest" 边界（取所有文件最小键中的最小值）
        if c.cmp(&f.smallest, smallest.as_ref().unwrap()) == Ordering::Less {
            smallest = Some(f.smallest.clone());
        }
        // 第四步：更新 "largest" 边界（取所有文件最大键中的最大值）
        if c.cmp(&f.largest, largest.as_ref().unwrap()) == Ordering::Greater {
            largest = Some(f.largest.clone());
        }
    }
    (smallest.unwrap(), largest.unwrap())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cache::Cache;
    use crate::cmp::DefaultCmp;
    use crate::key_types::LookupKey;
    use crate::test_util::LdbIteratorIter;
    use crate::types::FileMetaData;
    use crate::version::testutil::make_version;

    fn example_files() -> Vec<FileMetaHandle> {
        let f1 = FileMetaData {
            num: 1,
            size: 10,
            smallest: b"f".to_vec().into(),
            largest: b"g".to_vec().into(),
            ..Default::default()
        };
        let f2 = FileMetaData {
            num: 2,
            size: 20,
            smallest: b"e".to_vec().into(),
            largest: b"f".to_vec().into(),
            ..Default::default()
        };
        let f3 = FileMetaData {
            num: 3,
            size: 30,
            smallest: b"a".to_vec().into(),
            largest: b"b".to_vec().into(),
            ..Default::default()
        };
        let f4 = FileMetaData {
            num: 4,
            size: 40,
            smallest: b"q".to_vec().into(),
            largest: b"z".to_vec().into(),
            ..Default::default()
        };
        vec![f1, f2, f3, f4].into_iter().map(share).collect()
    }

    #[test]
    fn test_version_set_merge_iters() {
        let v1 = vec![2, 4, 6, 8, 10];
        let v2 = vec![1, 3, 5, 7];
        assert_eq!(
            vec![1, 2, 3, 4, 5, 6, 7, 8, 10],
            merge_iters(v1.into_iter(), v2.into_iter(), |a, b| a.cmp(b))
        );
    }

    #[test]
    fn test_version_set_total_size() {
        assert_eq!(100, total_size(example_files().iter()));
    }

    #[test]
    fn test_version_set_get_range() {
        let cmp = DefaultCmp;
        let fs = example_files();
        assert_eq!(
            (b"a".to_vec().into(), b"z".to_vec().into()),
            get_range(&cmp, fs.iter())
        );
    }

    #[test]
    fn test_version_set_builder() {
        let (v, opt) = make_version();
        let v = share(v);

        let fmd = FileMetaData {
            num: 21,
            size: 123,
            smallest: LookupKey::new(b"klm", 777).internal_key().to_vec().into(),
            largest: LookupKey::new(b"kop", 700).internal_key().to_vec().into(),
            ..Default::default()
        };

        let mut ve = VersionEdit::new();
        ve.add_file(1, fmd);
        // This deletion should be undone by apply().
        ve.delete_file(1, 21);
        ve.delete_file(0, 2);
        ve.set_compact_pointer(2, LookupKey::new(b"xxx", 123).internal_key());

        let mut b = Builder::new();
        let mut ptrs: [Bytes; NUM_LEVELS] = Default::default();
        b.apply(&ve, &mut ptrs);

        assert_eq!(
            &[120_u8, 120, 120, 1, 123, 0, 0, 0, 0, 0, 0],
            ptrs[2].as_ref()
        );
        assert_eq!(2, b.deleted[0][0]);
        assert_eq!(1, b.added[1].len());

        let mut v2 = Version::new(
            share(TableCache::new(
                "db",
                opt.clone(),
                share(Cache::new(128)),
                100,
            )),
            opt.cmp.clone(),
        );
        b.save_to(&InternalKeyCmp(opt.cmp.clone()), &v, &mut v2);
        // Second file in L0 was removed.
        assert_eq!(1, v2.files[0].len());
        // File was added to L1.
        assert_eq!(4, v2.files[1].len());
        assert_eq!(21, v2.files[1][3].borrow().num);
    }

    #[test]
    fn test_version_set_log_and_apply() {
        let (_, opt) = make_version();
        let mut vs = VersionSet::new(
            "db",
            opt.clone(),
            share(TableCache::new(
                "db",
                opt.clone(),
                share(Cache::new(128)),
                100,
            )),
        );

        assert_eq!(2, vs.new_file_number());
        // Simulate NewDB
        {
            let mut ve = VersionEdit::new();
            ve.set_comparator_name("leveldb.BytewiseComparator");
            ve.set_log_num(10);
            ve.set_next_file(20);
            ve.set_last_seq(30);

            // Write first manifest to be recovered from.
            let manifest = manifest_file_name("db", 19);
            let mffile = opt.env.open_writable_file(Path::new(&manifest)).unwrap();
            let mut lw = LogWriter::new(mffile);
            lw.add_record(&ve.encode()).unwrap();
            lw.flush().unwrap();
            set_current_file(opt.env.as_ref().as_ref(), "db", 19).unwrap();
        }

        // Recover from new state.
        {
            vs.recover().unwrap();
            assert_eq!(10, vs.log_num);
            assert_eq!(21, vs.next_file_num);
            assert_eq!(30, vs.last_seq);
            assert_eq!(0, vs.current.as_ref().unwrap().borrow().files[0].len());
            assert_eq!(0, vs.current.as_ref().unwrap().borrow().files[1].len());
            assert_eq!(35, vs.write_snapshot().unwrap());
        }

        // Simulate compaction by adding a file.
        {
            let mut ve = VersionEdit::new();
            ve.set_log_num(11);
            let fmd = FileMetaData {
                num: 21,
                size: 123,
                smallest: LookupKey::new(b"abc", 777).internal_key().to_vec().into(),
                largest: LookupKey::new(b"def", 700).internal_key().to_vec().into(),
                ..Default::default()
            };
            ve.add_file(1, fmd);
            vs.log_and_apply(ve).unwrap();

            assert!(opt.env.exists(&Path::new("db").join("CURRENT")).unwrap());
            assert!(opt
                .env
                .exists(&Path::new("db").join("MANIFEST-000019"))
                .unwrap());
            // next_file_num and last_seq are untouched by log_and_apply
            assert_eq!(21, vs.new_file_number());
            assert_eq!(22, vs.next_file_num);
            assert_eq!(30, vs.last_seq);
            // the following fields are touched by log_and_apply.
            assert_eq!(11, vs.log_num);

            // The previous "compaction" should have added one file to the first level in the
            // current version.
            assert_eq!(0, vs.current.as_ref().unwrap().borrow().files[0].len());
            assert_eq!(1, vs.current.as_ref().unwrap().borrow().files[1].len());
            assert_eq!(63, vs.write_snapshot().unwrap());
        }
    }

    #[test]
    fn test_version_set_utils() {
        let (v, opt) = make_version();
        let mut vs = VersionSet::new(
            "db",
            opt.clone(),
            share(TableCache::new("db", opt, share(Cache::new(128)), 100)),
        );
        vs.add_version(v);
        // live_files()
        assert_eq!(9, vs.live_files().len());
        assert!(vs.live_files().contains(&3));

        let v = vs.current();
        let v = v.borrow();
        // num_level_bytes()
        assert_eq!(483, v.num_level_bytes(0));
        assert_eq!(651, v.num_level_bytes(1));
        assert_eq!(468, v.num_level_bytes(2));
        // num_level_files()
        assert_eq!(2, v.num_level_files(0));
        assert_eq!(3, v.num_level_files(1));
        assert_eq!(2, v.num_level_files(2));
        // new_file_number()
        assert_eq!(2, vs.new_file_number());
        assert_eq!(3, vs.new_file_number());
    }

    #[test]
    fn test_version_set_pick_compaction() {
        let (mut v, opt) = make_version();
        let mut vs = VersionSet::new(
            "db",
            opt.clone(),
            share(TableCache::new("db", opt, share(Cache::new(128)), 100)),
        );

        v.compaction_score = Some(2.0);
        v.compaction_level = Some(0);
        vs.add_version(v);

        // Size compaction
        {
            let c = vs.pick_compaction().unwrap();
            assert_eq!(2, c.inputs[0].len());
            assert_eq!(1, c.inputs[1].len());
            assert_eq!(0, c.level);
            assert!(c.input_version.is_some());
        }
        // Seek compaction
        {
            let current = vs.current();
            current.borrow_mut().compaction_score = None;
            current.borrow_mut().compaction_level = None;
            current.borrow_mut().file_to_compact_lvl = 1;

            let fmd = current.borrow().files[1][0].clone();
            current.borrow_mut().file_to_compact = Some(fmd);

            let c = vs.pick_compaction().unwrap();
            assert_eq!(3, c.inputs[0].len()); // inputs on l+0 are expanded.
            assert_eq!(1, c.inputs[1].len());
            assert_eq!(1, c.level);
            assert!(c.input_version.is_some());
        }
    }

    /// iterator_properties tests that it contains len elements and that they are ordered in
    /// ascending order by cmp.
    fn iterator_properties<It: LdbIterator>(mut it: It, len: usize, cmp: Rc<Box<dyn Cmp>>) {
        let mut wr = LdbIteratorIter::wrap(&mut it);
        let first = wr.next().unwrap();
        let mut count = 1;
        wr.fold(first, |(a, _), (b, c)| {
            assert_eq!(Ordering::Less, cmp.cmp(&a, &b));
            count += 1;
            (b, c)
        });
        assert_eq!(len, count);
    }

    #[test]
    fn test_version_set_compaction() {
        let (v, opt) = make_version();
        let mut vs = VersionSet::new(
            "db",
            opt.clone(),
            share(TableCache::new("db", opt, share(Cache::new(128)), 100)),
        );
        time_test!();
        vs.add_version(v);

        {
            // approximate_offset()
            let v = vs.current();
            assert_eq!(
                0,
                vs.approximate_offset(&v, LookupKey::new(b"aaa", 9000).internal_key())
            );
            assert_eq!(
                232,
                vs.approximate_offset(&v, LookupKey::new(b"bab", 9000).internal_key())
            );
            assert_eq!(
                1134,
                vs.approximate_offset(&v, LookupKey::new(b"fab", 9000).internal_key())
            );
        }
        // The following tests reuse the same version set and verify that various compactions work
        // like they should.
        {
            time_test!("compaction tests");
            // compact level 0 with a partial range.
            let from = LookupKey::new(b"000", 1000);
            let to = LookupKey::new(b"ab", 1010);
            let c = vs
                .compact_range(0, from.internal_key(), to.internal_key())
                .unwrap();
            assert_eq!(2, c.inputs[0].len());
            assert_eq!(1, c.inputs[1].len());
            assert_eq!(1, c.grandparents.unwrap().len());

            // compact level 0, but entire range of keys in version.
            let from = LookupKey::new(b"000", 1000);
            let to = LookupKey::new(b"zzz", 1010);
            let c = vs
                .compact_range(0, from.internal_key(), to.internal_key())
                .unwrap();
            assert_eq!(2, c.inputs[0].len());
            assert_eq!(1, c.inputs[1].len());
            assert_eq!(1, c.grandparents.as_ref().unwrap().len());
            iterator_properties(
                vs.make_input_iterator(&c),
                12,
                Rc::new(Box::new(vs.cmp.clone())),
            );

            // Expand input range on higher level.
            let from = LookupKey::new(b"dab", 1000);
            let to = LookupKey::new(b"eab", 1010);
            let c = vs
                .compact_range(1, from.internal_key(), to.internal_key())
                .unwrap();
            assert_eq!(3, c.inputs[0].len());
            assert_eq!(1, c.inputs[1].len());
            assert_eq!(0, c.grandparents.as_ref().unwrap().len());
            iterator_properties(
                vs.make_input_iterator(&c),
                12,
                Rc::new(Box::new(vs.cmp.clone())),
            );

            // is_trivial_move
            let from = LookupKey::new(b"fab", 1000);
            let to = LookupKey::new(b"fba", 1010);
            let mut c = vs
                .compact_range(2, from.internal_key(), to.internal_key())
                .unwrap();
            // pretend it's not manual
            c.manual = false;
            assert!(c.is_trivial_move());

            // should_stop_before
            let from = LookupKey::new(b"000", 1000);
            let to = LookupKey::new(b"zzz", 1010);
            let mid = LookupKey::new(b"abc", 1010);
            let mut c = vs
                .compact_range(0, from.internal_key(), to.internal_key())
                .unwrap();
            assert!(!c.should_stop_before(from.internal_key()));
            assert!(!c.should_stop_before(mid.internal_key()));
            assert!(!c.should_stop_before(to.internal_key()));

            // is_base_level_for
            let from = LookupKey::new(b"000", 1000);
            let to = LookupKey::new(b"zzz", 1010);
            let mut c = vs
                .compact_range(0, from.internal_key(), to.internal_key())
                .unwrap();
            assert!(c.is_base_level_for(b"aaa"));
            assert!(!c.is_base_level_for(b"hac"));

            // input/add_input_deletions
            let from = LookupKey::new(b"000", 1000);
            let to = LookupKey::new(b"zzz", 1010);
            let mut c = vs
                .compact_range(0, from.internal_key(), to.internal_key())
                .unwrap();
            for inp in &[(0, 0, 1), (0, 1, 2), (1, 0, 3)] {
                let f = &c.inputs[inp.0][inp.1];
                assert_eq!(inp.2, f.borrow().num);
            }
            c.add_input_deletions();
            assert_eq!(23, c.edit().encode().len())
        }
    }
}
